93e13c03995e0b7dd19d32150d2508ba556e2e0e,src/streambench/flinkbench/src/main/java/com/intel/flinkbench/microbench/Split.java,Split,processStream,#FlinkBenchConfig#,31

Before Change


        final String seperator = config.separator;
        createDataStream(config);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, String>> dataStream = getDataStream();
        dataStream.flatMap(new FlatMapFunction<Tuple2<String, String>, Tuple2<String, String>>() {
            @Override
            public void flatMap(Tuple2<String, String> value, Collector<Tuple2<String, String>> out)

After Change


        final String seperator = config.separator;
        createDataStream(config);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        DataStream<Tuple2<String, String>> dataStream = env.addSource(getDataStream());
        dataStream.flatMap(new FlatMapFunction<Tuple2<String, String>, Tuple2<String, String>>() {
            @Override
            public void flatMap(Tuple2<String, String> value, Collector<Tuple2<String, String>> out)